package org.ligson.spark.shuangseqiu;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.mllib.stat.MultivariateStatisticalSummary;
import org.apache.spark.mllib.stat.Statistics;
import org.ligson.spark.util.SparkContextUtil;
import scala.Function1;
import scala.Tuple2;

import java.util.*;

/***
 * 统计每列上出现最少的数字
 */
public class CoutDemo1 {
    public static void main(String[] args) {
        JavaSparkContext context = SparkContextUtil.getLocalContext();
        JavaRDD<String[]> data = context.textFile("./src/main/resources/newestssq.txt").map(new Function<String, String[]>() {
            public String[] call(String val) throws Exception {
                String[] array = val.split("\t");
                String[] array2 = new String[array.length - 2];
                System.arraycopy(array, 2, array2, 0, array2.length);
                return array2;
            }
        });
        JavaRDD<Vector> data2 = data.map(new Function<String[], Vector>() {
            public Vector call(String[] v1) throws Exception {
                double[] doubles = new double[v1.length];
                int i = 0;
                for (String s : v1) {
                    doubles[i++] = Double.parseDouble(s);
                }
                return Vectors.dense(doubles);
            }
        });
        MultivariateStatisticalSummary summ = Statistics.colStats(data2.rdd());
        System.out.println(summ.max());
        System.out.println(summ.min());
        System.out.println(summ.mean());
        JavaPairRDD<Double, Integer> data3 = data2.flatMapToPair(new PairFlatMapFunction<Vector, Double, Integer>() {
            public Iterator<Tuple2<Double, Integer>> call(Vector vector) throws Exception {
                double[] doubles = vector.toArray();
                List<Tuple2<Double, Integer>> list = new ArrayList<Tuple2<Double, Integer>>();
                for (double aDouble : doubles) {
                    list.add(Tuple2.apply(aDouble, 1));
                }
                System.out.println(list);
                return list.iterator();
            }
        });
        JavaPairRDD<Double, Integer> result = data3.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        result.foreach(new VoidFunction<Tuple2<Double, Integer>>() {
            public void call(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                System.out.println(doubleIntegerTuple2._1 + "==" + doubleIntegerTuple2._2);
            }
        });
        System.out.println(result.rdd().sortBy$default$2());
        result.foreach(new VoidFunction<Tuple2<Double, Integer>>() {
            public void call(Tuple2<Double, Integer> doubleIntegerTuple2) throws Exception {
                System.out.println(doubleIntegerTuple2._1 + "==" + doubleIntegerTuple2._2);
            }
        });

    }
}
